package kube

import (
	"fmt"
	"net"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/pkg/errors"
	bkeaddon "gopkg.openfuyao.cn/bkecommon/cluster/addon"
	confv1beta1 "gopkg.openfuyao.cn/bkecommon/cluster/api/v1beta1"
	bkeinit "gopkg.openfuyao.cn/bkecommon/cluster/initialize"
	bkenode "gopkg.openfuyao.cn/bkecommon/cluster/node"
	bkenet "gopkg.openfuyao.cn/bkecommon/utils/net"
	apierrors2 "k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/util/json"

	bkev1beta1 "gopkg.openfuyao.cn/cluster-api-provider-bke/api/v1beta1"
	metricrecord "gopkg.openfuyao.cn/cluster-api-provider-bke/pkg/metrics/record"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils/constant"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils/log"
	v "gopkg.openfuyao.cn/cluster-api-provider-bke/version"
)

const (
	keyValueSeparatorCount     = 2  // for parseNodeSelector key=value splitting
	bocOperatorTimeoutMinutes  = 30 // timeout for bocoperator addon in minutes
	bocOperatorIntervalSeconds = 30 // interval for bocoperator addon in seconds
)

type Task struct {
	Name     string
	FilePath string
	Param    map[string]interface{}
	// IgnoreError: if true, continue to apply the remaining yaml ignoring errors
	IgnoreError bool
	// Block: if true, wait for the task to complete
	Block    bool
	Timeout  time.Duration
	Interval time.Duration
	Operate  bkeaddon.AddonOperate
	recorder *AddonRecorder
}

type AddonRecorder struct {
	AddonName    string
	AddonVersion string
	AddonObjects []*AddonObject
}

type AddonObject struct {
	Name      string
	Kind      string
	NameSpace string
}

func NewAddonRecorder(addonT *bkeaddon.AddonTransfer) *AddonRecorder {
	return &AddonRecorder{
		AddonName:    addonT.Addon.Name,
		AddonVersion: addonT.Addon.Version,
		AddonObjects: []*AddonObject{},
	}
}

func (recorder *AddonRecorder) Record(obj *unstructured.Unstructured) {
	recorder.AddonObjects = append(recorder.AddonObjects, NewAddonObject(obj))
}

func NewAddonObject(obj *unstructured.Unstructured) *AddonObject {
	return &AddonObject{
		Name:      obj.GetName(),
		Kind:      obj.GetKind(),
		NameSpace: obj.GetNamespace(),
	}
}

func NewTask(name, filePath string, param map[string]interface{}) *Task {
	return &Task{
		Name:        name,
		Param:       param,
		FilePath:    filePath,
		IgnoreError: false,
		Block:       true,
	}
}

func (t *Task) SetWaiter(wait bool, timeout time.Duration, interval time.Duration) *Task {
	t.Block = wait
	t.Timeout = timeout
	t.Interval = interval
	return t
}

func (t *Task) AddRepo(repo string) *Task {
	if t.Param == nil {
		t.Param = make(map[string]interface{})
	}
	t.Param["repo"] = repo
	return t
}

func (t *Task) SetOperate(operate bkeaddon.AddonOperate) *Task {
	t.Operate = operate
	return t
}

func (t *Task) RegisAddonRecorder(recorder *AddonRecorder) *Task {
	t.recorder = recorder
	return t
}

func (c *Client) InstallAddon(bkeCluster *bkev1beta1.BKECluster, addonT *bkeaddon.AddonTransfer, addonRecorder *AddonRecorder) error {
	var err error
	defer metricrecord.AddonInstallRecord(bkeCluster, addonT.Addon.Name, addonT.Addon.Version, err)()

	addon := addonT.Addon.DeepCopy()
	bkeConfig := bkeCluster.Spec.ClusterConfig
	cfg := bkeinit.BkeConfig(*bkeConfig)
	repo := cfg.ImageRepo()

	files, err := c.getAddonYamlFiles(addon)
	if err != nil {
		return err
	}

	if addonT.Operate == bkeaddon.RemoveAddon {
		sort.Sort(sort.Reverse(sort.StringSlice(files)))
	} else {
		sort.Strings(files)
	}

	param, err := c.prepareAddonParameters(bkeCluster, addon, files, repo)
	if err != nil {
		return err
	}

	applyConfig := &addonApplyConfig{
		addon:         addon,
		addonT:        addonT,
		files:         files,
		param:         param,
		repo:          repo,
		addonRecorder: addonRecorder,
	}

	return c.applyAddonFiles(applyConfig)
}

func (c *Client) getAddonYamlFiles(addon *confv1beta1.Product) ([]string, error) {
	addonDir := filepath.Join(constant.K8sManifestsDir, addon.Name, addon.Version)
	if _, err := os.Stat(addonDir); os.IsNotExist(err) {
		return nil, errors.Errorf("addon dir %q not exist,err: %v", addonDir, err)
	}

	var files []string
	err := filepath.Walk(addonDir, func(path string, info os.FileInfo, err error) error {
		if info.IsDir() {
			return nil
		}
		if strings.HasSuffix(path, ".yaml") {
			files = append(files, path)
		}
		return nil
	})
	if err != nil {
		return nil, errors.Wrapf(err, "failed to traverse addon dir %q", addonDir)
	}
	if len(files) == 0 {
		return nil, errors.Errorf("no yaml file found in dir %q", addonDir)
	}
	return files, nil
}

func (c *Client) prepareAddonParameters(bkeCluster *bkev1beta1.BKECluster, addon *confv1beta1.Product, files []string, repo string) (map[string]map[string]interface{}, error) {
	commonParam, err := getCommonParamFromBKECluster(bkeCluster)
	if err != nil {
		return nil, err
	}

	commonParam, err = c.enhanceCommonParamForSpecialAddons(bkeCluster, addon, commonParam)
	if err != nil {
		return nil, err
	}

	filesBaseNames := make([]string, len(files))
	for i, file := range files {
		filesBaseNames[i] = strings.Split(filepath.Base(file), ".")[0]
	}

	param := prepareAddonParam(addon.Param, filesBaseNames, repo)
	if err := normalizeNodeSelector(param); err != nil {
		return nil, err
	}

	for i := range files {
		fileName := filesBaseNames[i]
		param[fileName] = mergeParam(commonParam, param[fileName])
	}

	return param, nil
}

func (c *Client) enhanceCommonParamForSpecialAddons(bkeCluster *bkev1beta1.BKECluster, addon *confv1beta1.Product, commonParam map[string]interface{}) (map[string]interface{}, error) {
	if commonParam == nil {
		commonParam = make(map[string]interface{})
	}

	if addon.Name == "bocoperator" {
		bocParam, err := getBocDefaultParam(bkeCluster, addon)
		if err != nil {
			return nil, err
		}

		portalClusterToken, err := c.getPortalK8sToken()
		if err != nil {
			return nil, err
		}
		bocParam["portalClusterToken"] = portalClusterToken
		commonParam = mergeParam(commonParam, bocParam)
	}

	if addon.Name == "cluster-api" && addon.Param["manage"] == "true" {
		portalClusterToken, err := c.getPortalK8sToken()
		if err != nil {
			return nil, err
		}
		commonParam["clusterToken"] = portalClusterToken
	}

	if addon.Name == "fabric" {
		fabricParam, err := parseFabricParam(addon)
		if err != nil {
			return nil, err
		}
		addon.Param = fabricParam
	}

	return commonParam, nil
}

type addonApplyConfig struct {
	addon         *confv1beta1.Product
	addonT        *bkeaddon.AddonTransfer
	files         []string
	param         map[string]map[string]interface{}
	repo          string
	addonRecorder *AddonRecorder
}

func (c *Client) applyAddonFiles(config *addonApplyConfig) error {
	filesBaseNames := c.extractFileBaseNames(config.files)

	for i, file := range config.files {
		fileName := filesBaseNames[i]
		task := c.createAddonTask(config, fileName, file)

		if err := c.ApplyYaml(task); err != nil {
			return c.handleApplyError(err, config.addon, file, config.addonT.Operate)
		}
	}
	return nil
}

func (c *Client) extractFileBaseNames(files []string) []string {
	filesBaseNames := make([]string, len(files))
	for i, file := range files {
		filesBaseNames[i] = strings.Split(filepath.Base(file), ".")[0]
	}
	return filesBaseNames
}

func (c *Client) createAddonTask(config *addonApplyConfig, fileName, file string) *Task {
	task := NewTask(config.addon.Name, file, config.param[fileName]).
		AddRepo(config.repo).
		SetWaiter(config.addon.Block, bkeinit.DefaultAddonTimeout, bkeinit.DefaultAddonInterval).
		SetOperate(config.addonT.Operate).
		RegisAddonRecorder(config.addonRecorder)

	if config.addonT.Operate == bkeaddon.CreateAddon && config.addon.Name == "bocoperator" {
		task.SetWaiter(config.addon.Block, bocOperatorTimeoutMinutes*time.Minute, bocOperatorIntervalSeconds*time.Second)
	}

	return task
}

func (c *Client) handleApplyError(err error, addon *confv1beta1.Product, file string, operate bkeaddon.AddonOperate) error {
	if operate == bkeaddon.RemoveAddon {
		c.Log.Warnf("(ignore)failed to remove addon %s/%s, err: %v", addon.Name, addon.Version, err)
		return nil
	}

	if apierrors2.IsNoMatchError(err) {
		c.Log.Errorf("The addon %s/%s yaml file contains resources that are not supported in the target cluster，err: %v", addon.Name, addon.Version, err)
	} else {
		c.Log.Warnf("failed to apply yaml %q, err: %v", file, err)
	}
	return err
}

func (c *Client) getPortalK8sToken() (string, error) {
	return c.NewK8sToken()
}

func getBocDefaultParam(bkeCluster *bkev1beta1.BKECluster, addon *confv1beta1.Product) (map[string]interface{}, error) {
	param := make(map[string]interface{})

	bkeConfig := bkeinit.BkeConfig(*bkeCluster.Spec.ClusterConfig)

	bkeNodes := bkenode.Nodes(bkeConfig.Nodes)

	masterNodes := bkeNodes.Master().Decrypt()
	param["masterSshIp"] = masterNodes[0].IP
	param["masterSshPort"] = masterNodes[0].Port
	param["masterSshUser"] = masterNodes[0].Username
	param["masterSshPwd"] = masterNodes[0].Password

	param["pipeLineServerIp"] = masterNodes[0].IP
	param["pipeLineServerPort"] = masterNodes[0].Port
	param["pipeLineServerUser"] = masterNodes[0].Username
	param["pipeLineServerPwd"] = masterNodes[0].Password
	param["pipeLineServerHostname"] = masterNodes[0].Hostname

	if ip, ok := bkeConfig.CustomExtra["pipelineServer"]; ok && ip != "" {
		if !bkenet.ValidIP(ip) {
			return nil, errors.Errorf("invalid pipelineServer IP %q", ip)
		}
		pipeLineNodes := masterNodes.Filter(bkenode.FilterOptions{"IP": ip})
		if len(pipeLineNodes) == 0 || len(pipeLineNodes) != 1 {
			log.Warnf("invalid pipelineServer IP %q, use master 0", ip)
		} else {
			param["pipeLineServerIp"] = pipeLineNodes[0].IP
			param["pipeLineServerPort"] = pipeLineNodes[0].Port
			param["pipeLineServerUser"] = pipeLineNodes[0].Username
			param["pipeLineServerPwd"] = pipeLineNodes[0].Password
			param["pipeLineServerHostname"] = pipeLineNodes[0].Hostname
		}
	}

	param["isLocalDB"] = "true"
	param["bocVersion"] = "v4.0"

	masterServersJson, err := json.Marshal(masterNodes)
	if err != nil {
		return nil, err
	}
	param["masterServers"] = string(masterServersJson)

	param["dbHost"] = bkeCluster.Spec.ControlPlaneEndpoint.Host

	tmp := make(map[string]interface{})
	for k, v := range addon.Param {
		tmp[k] = v
	}
	return mergeParam(param, tmp), nil
}

func getCommonParamFromBKECluster(bkeCluster *bkev1beta1.BKECluster) (map[string]interface{}, error) {
	param := make(map[string]interface{})
	param["replicas"] = 1
	param["kubeConfigDir"] = "/etc/kubernetes"
	param["ipModVersion"] = "ipv4"
	param["apiServerSrcPort"] = bkeinit.DefaultAPIBindPort
	param["podSubnet"] = bkeinit.DefaultPodSubnet
	param["serviceSubnet"] = bkeinit.DefaultServicesSubnet
	param["dnsIP"] = bkeinit.DefaultClusterDNSIP
	param["dnsDomain"] = bkeinit.DefaultServiceDNSDomain
	param["ingressReplicas"] = 1
	param["etcdEndpoints"] = ""
	param["version"] = v.Version
	param["kubeletDataRoot"] = bkeinit.DefaultKubeletRootDir
	param["dockerDataRoot"] = bkeinit.DefaultCRIDockerDataRootDir
	param["k8sVersion"] = bkeinit.DefaultKubernetesVersion
	param["masterReplicas"] = "1"
	param["workerReplicas"] = "1"
	param["namespace"] = "cluster-system"
	param["clusterNetworkMode"] = ""
	param["product"] = "boc"

	if bkeCluster == nil {
		return param, nil
	}

	if bkeCluster.Spec.ClusterConfig == nil {
		return param, nil
	}
	bkeConfig := bkeinit.BkeConfig(*bkeCluster.Spec.ClusterConfig)

	param["imageRepo"] = fmt.Sprintf("%s:%s", bkeConfig.Cluster.ImageRepo.Domain, bkeConfig.Cluster.ImageRepo.Port)
	param["imageRepoDomain"] = bkeConfig.Cluster.ImageRepo.Domain
	param["imageRepoPort"] = bkeConfig.Cluster.ImageRepo.Port
	param["imageRepoIp"] = bkeConfig.Cluster.ImageRepo.Ip
	param["imageRepoPrefix"] = bkeConfig.Cluster.ImageRepo.Prefix

	param["httpRepo"] = bkeConfig.YumRepo()
	param["httpRepoDomain"] = bkeConfig.Cluster.HTTPRepo.Domain
	param["httpRepoPort"] = bkeConfig.Cluster.HTTPRepo.Port
	param["httpRepoIp"] = bkeConfig.Cluster.HTTPRepo.Ip
	param["httpRepoPrefix"] = bkeConfig.Cluster.HTTPRepo.Prefix

	param["ntpServer"] = bkeConfig.Cluster.NTPServer

	param["nfsServer"] = bkeCluster.Spec.ClusterConfig.CustomExtra["nfsServer"]
	param["nfsRootDir"] = bkeCluster.Spec.ClusterConfig.CustomExtra["nfsRootDir"]
	param["nfsVersion"] = bkeCluster.Spec.ClusterConfig.CustomExtra["nfsVersion"]

	param["namespace"] = bkeCluster.Namespace

	if bkeConfig.Cluster.Networking.PodSubnet != "" {
		param["podSubnet"] = bkeConfig.Cluster.Networking.PodSubnet
	}
	if bkeConfig.Cluster.Networking.ServiceSubnet != "" {
		param["serviceSubnet"] = bkeConfig.Cluster.Networking.ServiceSubnet
	}
	if bkeConfig.Cluster.Networking.DNSDomain != "" {
		param["dnsDomain"] = bkeConfig.Cluster.Networking.DNSDomain
	}
	if bkeCluster.Spec.ControlPlaneEndpoint.Host != "" {
		param["apiServerSrcHost"] = bkeCluster.Spec.ControlPlaneEndpoint.Host
	}
	if bkeCluster.Spec.ControlPlaneEndpoint.Port != 0 {
		param["apiServerSrcPort"] = bkeCluster.Spec.ControlPlaneEndpoint.Port
	}
	bkeNodes := bkenode.Nodes(bkeConfig.Nodes)
	etcdEndpointsLi := []string{}
	for _, node := range bkeNodes.Etcd() {
		tmp := fmt.Sprintf("https://%s:2379", node.IP)
		etcdEndpointsLi = append(etcdEndpointsLi, tmp)
	}
	param["etcdEndpoints"] = strings.Join(etcdEndpointsLi, ",")

	param["masterReplicas"] = strconv.Itoa(bkeNodes.Master().Length())
	param["workerReplicas"] = strconv.Itoa(bkeNodes.Worker().Length())

	// for beyondelb replicas
	for _, v := range bkeConfig.Addons {
		name := strings.ToLower(v.Name)
		if name == "beyondelb" {
			lbNodes := v.Param["lbNodes"]
			param["ingressReplicas"] = len(strings.Split(lbNodes, ","))
		}
		if name == "calico" {
			param["clusterNetworkMode"] = "calico"
		}
		if name == "fabric" {
			param["clusterNetworkMode"] = "fabric"
		}
	}
	dnsIP, err := bkeinit.GetClusterDNSIP(param["serviceSubnet"].(string))
	if err != nil {
		return nil, errors.New("failed to get cluster dns ip")
	}
	param["dnsIP"] = dnsIP

	etcdNodes := bkenode.Nodes(bkeConfig.Nodes).Etcd()
	etcdEndpoints := []string{}
	for _, node := range etcdNodes {
		etcdEndpoints = append(etcdEndpoints, node.IP)
	}
	param["etcdIps"] = strings.Join(etcdEndpoints, ",")

	if bkeConfig.Cluster.Kubelet != nil && bkeConfig.Cluster.Kubelet.ExtraVolumes != nil {
		for _, volume := range bkeConfig.Cluster.Kubelet.ExtraVolumes {
			if volume.Name == "kubelet-root-dir" {
				param["kubeletDataRoot"] = volume.HostPath
			}
		}
	}

	if bkeConfig.Cluster.ContainerRuntime.CRI == bkeinit.CRIDocker {
		if v, ok := bkeConfig.Cluster.ContainerRuntime.Param["data-root"]; ok {
			param["dockerDataRoot"] = v
		}
	}

	if bkeConfig.Cluster.KubernetesVersion != "" {
		param["k8sVersion"] = bkeConfig.Cluster.KubernetesVersion
	}

	return param, nil
}

// normalizeNodeSelector normalize nodeSelector param from string to map[string]string
func normalizeNodeSelector(param map[string]map[string]interface{}) error {
	for file, kv := range param {
		raw, ok := kv["nodeSelector"]
		if !ok || raw == nil {
			continue
		}
		switch v := raw.(type) {
		case map[string]string, map[string]interface{}:
			continue
		case string:
			selector, err := parseNodeSelector(v)
			if err != nil {
				return errors.Wrapf(err, "%s.nodeSelector", file)
			}
			kv["nodeSelector"] = selector
		default:
			return errors.Errorf("%s.nodeSelector unsupported type %T", file, v)
		}
	}
	return nil
}

// parseNodeSelector parse nodeSelector param from string to map[string]string
func parseNodeSelector(raw string) (map[string]string, error) {
	out := make(map[string]string)
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return out, nil
	}
	if strings.HasPrefix(raw, "{") {
		if err := json.Unmarshal([]byte(raw), &out); err != nil {
			return nil, err
		}
		return out, nil
	}
	for _, pair := range strings.Split(raw, ",") {
		pair = strings.TrimSpace(pair)
		if pair == "" {
			continue
		}
		kv := strings.SplitN(pair, "=", keyValueSeparatorCount)
		key := strings.TrimSpace(kv[0])
		if key == "" {
			return nil, errors.Errorf("invalid nodeSelector entry %q", pair)
		}
		val := ""
		if len(kv) == keyValueSeparatorCount {
			val = strings.TrimSpace(kv[1])
		}
		out[key] = val
	}
	return out, nil
}

// parseFabricParam parse fabric param excludeIps
func parseFabricParam(addon *confv1beta1.Product) (map[string]string, error) {
	fabricParam := addon.Param

	if p, ok := fabricParam["excludeIps"]; ok && p != "" {
		items := strings.Split(p, ",")
		excludeIps := []string{}
		for _, item := range items {
			if strings.Contains(item, "-") {
				ips, err := parseFabricExcludeIPRange(item)
				if err != nil {
					return nil, err
				}
				excludeIps = append(excludeIps, ips...)
				continue
			}

			if ip := net.ParseIP(item); ip != nil {
				excludeIps = append(excludeIps, ip.String())
			} else {
				return nil, errors.Errorf("fabric param excludeIps contain a invalid ip %s", item)
			}
		}
		fabricParam["excludeIps"] = strings.Join(excludeIps, ",")
		return fabricParam, nil
	}
	return fabricParam, nil
}

func mergeParam(src, dst map[string]interface{}) map[string]interface{} {
	for k, v := range src {
		if _, ok := dst[k]; !ok {
			dst[k] = v
		}
	}
	return dst
}

// prepareAddonParam prepare addon param, multi yaml file and single yaml file param format
// before ->
//
//	param = map[string]string{
//			"filename1.key1": value1,
//			"filename2.key2": value2,
//	     }
//
//	or ->  param = map[string]string{
//					"key1": value1,
//					"key2": value2,
//	             }
//
// invalid param
//
//	param = map[string]string{
//			"key1": value1,
//			"filename1.key2": value2,
//	     }
//
// after ->
//
//	param = map[string]map[string]string{
//			"filename1": map[string]string{
//	                 "repo": "deploy.bocloud.k8s:40443/kubernetes/"
//	                 "key1": value1,
//	                 },
//			"filename2": map[string]string{
//	                 "repo": "deploy.bocloud.k8s:40443/kubernetes/"
//	                 "key2": value2,
//	                 },
//	}
func prepareAddonParam(addonParam map[string]string, filesBaseNames []string, repo string) map[string]map[string]interface{} {
	param := make(map[string]map[string]interface{})
	for _, fileName := range filesBaseNames {
		param[fileName] = make(map[string]interface{})
	}
	for k, v := range addonParam {
		arg := strings.Split(k, ".")
		if len(arg) == 1 {
			for _, fileName := range filesBaseNames {
				param[fileName][k] = v
			}
		} else {
			if param[arg[0]] == nil {
				param[arg[0]] = make(map[string]interface{})
			}
			param[arg[0]][arg[1]] = v
		}
	}
	// add default param repo
	for _, v := range param {
		if _, ok := v["repo"]; !ok {
			v["repo"] = repo
		}
	}
	return param
}

func parseFabricExcludeIPRange(rangeStr string) ([]string, error) {
	ipRange := strings.Split(rangeStr, "-")
	if len(ipRange) != 2 || ipRange == nil {
		return nil, errors.Errorf("fabric param excludeIps contain a invalid ip range %s", rangeStr)
	}
	start := net.ParseIP(ipRange[0]).To4()
	end := net.ParseIP(ipRange[1]).To4()
	if start == nil || end == nil {
		return nil, errors.Errorf("fabric param excludeIps contain a invalid ip range %s", rangeStr)
	}

	ipLe := func(src, dst net.IP) bool {
		if src.Equal(dst) {
			return false
		}
		for i, value := range src {
			if value > dst[i] {
				return true
			}
		}
		return false
	}

	if ipLe(start, end) {
		tmp := start
		start = end
		end = tmp
	}

	var ips []string
	ips = append(ips, start.String())
	for ip := start; !ip.Equal(end); {
		for j := len(ip) - 1; j >= 0; j-- {
			ip[j]++
			if ip[j] > 0 {
				break
			}
		}
		ips = append(ips, ip.String())
	}
	if !utils.ContainsString(ips, end.String()) {
		ips = append(ips, end.String())
	}
	return ips, nil
}
